package com.lfs.market.controller.temp;

import cn.hutool.json.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;

@Slf4j
public class CanalMain {
    public static void main(String[] args) {
        // 本地机器的地址
        String host = "192.168.187.135";
        int port = 11111;
        String destination = "example";
        // canal的账号
        String username = "canal_username";
        String password = "canal_username";

        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(host, port), destination, username, password);

        int batchSize = 100;

        while (true) {
            try {
                // 连接 Canal Server
                connector.connect();
                // 订阅表 remote.*
                connector.subscribe("remote\\.remote_user");
                // 回滚到上次位置
                connector.rollback();

                log.info("Canal 客户端启动成功，开始监听...");

                while (true) {
                    Message message = connector.getWithoutAck(batchSize);
                    long batchId = message.getId();
                    int size = message.getEntries().size();

                    if (batchId == -1 || size == 0) {
                        log.info("目前没有数据，等待...");
                        Thread.sleep(3000);
                        continue;
                    }

                    List<CanalEntry.Entry> entries = message.getEntries();
                    for (CanalEntry.Entry entry : entries) {
                        processEntry(entry);
                    }
                    // 手动确认消费完成
                    connector.ack(batchId);
                }
            } catch (Exception e) {
                log.error("Canal 客户端异常，准备重连...", e);
                try {
                    Thread.sleep(5000); // 失败后等待 5 秒再重连
                } catch (InterruptedException ignored) {
                }
            } finally {
                connector.disconnect();
                log.warn("Canal 客户端已断开连接");
            }
        }
    }

    private static void processEntry(CanalEntry.Entry entry) {
        try {
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                return;
            }

            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            String tableName = entry.getHeader().getTableName();
            CanalEntry.EventType eventType = rowChange.getEventType();
            List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();

            for (CanalEntry.RowData rowData : rowDataList) {
                JSONObject beforeData = new JSONObject();
                rowData.getBeforeColumnsList().forEach(column ->
                        beforeData.set(column.getName(), column.getValue()));

                JSONObject afterData = new JSONObject();
                rowData.getAfterColumnsList().forEach(column ->
                        afterData.set(column.getName(), column.getValue()));

                log.info("表名: {}, 事件类型: {}, 修改前: {}, 修改后: {}",
                        tableName, eventType, beforeData, afterData);
            }

        } catch (Exception e) {
            log.error("解析 binlog 数据失败", e);
        }
    }
}